package com.qupeng.demo.kafka.kafkaspringbootproducer.admin;

import com.qupeng.demo.kafka.kafkaspringbootproducer.config.KafkaProperties;
import com.qupeng.demo.kafka.kafkaspringbootproducer.model.CommonResponse;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ListOffsetsResult;
import org.apache.kafka.clients.admin.OffsetSpec;
import org.apache.kafka.clients.admin.RecordsToDelete;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.web.bind.annotation.DeleteMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;

@RestController
@RequestMapping("/admin/sync")
public class RecordSyncAdminController {

    private static final Logger LOGGER = LoggerFactory.getLogger(PartitionSyncAdminController.class);

    @Autowired
    private KafkaAdmin kafkaAdmin;

    @Autowired
    private AdminClient adminClient;

    @Autowired
    private KafkaProperties kafkaProperties;

    @DeleteMapping("/consumer-group/{name}/records")
    public CommonResponse<String> deleteRecords(@PathVariable String name) {
        try {
            Map<TopicPartition, OffsetAndMetadata> offsets = adminClient.listConsumerGroupOffsets(name).partitionsToOffsetAndMetadata().get();
            Map<TopicPartition, OffsetSpec> requestTimeOffsets = new HashMap<>();
            for (TopicPartition tp : offsets.keySet()) {
                requestTimeOffsets.put(tp, OffsetSpec.forTimestamp(System.currentTimeMillis() - 600*1000));
            }

            Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> olderOffsets = adminClient.listOffsets(requestTimeOffsets).all().get();
            Map<TopicPartition, RecordsToDelete> recordsToDelete = new HashMap<>();
            for (Map.Entry<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> entry : olderOffsets.entrySet()) {
                recordsToDelete.put(entry.getKey(), RecordsToDelete.beforeOffset(entry.getValue().offset()));
            }
            adminClient.deleteRecords(recordsToDelete).all();
            //adminClient.deleteRecords(recordsToDelete).lowWatermarks();
            return new CommonResponse<>("kafka-admin");
        }  catch (InterruptedException | ExecutionException ex) {
            return new CommonResponse<>("admin-consumer-group-003", ex.getMessage(), "kafka-admin");
        }
    }
}
